package cn.sheep.dolphin.report

import cn.sheep.dolphin.common.{DolphinAppComm, DolphinKpi}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import scalikejdbc._


/**
  * 媒体指标分析
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/12/1
  */
object AppAnalysisCore {

	def main(args: Array[String]): Unit = {

		if (args.size != 2) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.report.AppAnalysisCore
				  |	<parquetInputPath> <appdictFilePath>
				""".stripMargin)
			sys.exit(-1)
		}

		val Array(parquetInputPath, appdictFilePath) = args

		val sc = DolphinAppComm.createSparkContext("媒体指标分析")

		// 广播字典数据(全量数据)
		val appdictDataMap = sc.textFile(appdictFilePath)
			.map(line => line.split("\t", -1))
			.filter(_.size >= 5)
			.map(arr => (arr(4), arr(1))).collectAsMap()
		val appdictBt = sc.broadcast(appdictDataMap)


		val sQLContext = new SQLContext(sc)
		// 读取parquet文件
		val dataFrame = sQLContext.read.parquet(parquetInputPath)
		/*.filter("appid!=null or appid!='' or appname!=null")*/

		// 整理数据(K, V).reduceByKey
		dataFrame.map(row => {
			val day = row.getAs[String]("requestdate").substring(0, 10)

			var appName = row.getAs[String]("appname")
			if (StringUtils.isEmpty(appName)) {
				// 如果appName为空则根据appid去字典文件中找appName
				val appId = row.getAs[String]("appid")
				if (StringUtils.isNotEmpty(appId)) {
					appName = appdictBt.value.getOrElse(appId, appId)
				} else appName = "未知"
			}

			((day, appName), DolphinKpi.caculateKpi(row))
		})
			.reduceByKey((list1, list2) => list1 zip list2 map (tp => tp._1 + tp._2))
			.foreachPartition(partition => {
				DB localTx { implicit session =>

					/**
					  * 封装参数
					  * Seq(
					  * Seq(20181201,appName, rawReq,.........................) 第一行数据
					  * Seq(20181201,appName, rawReq,.........................) 第二行数据
					  * Seq(20181201,appName, rawReq,.........................) 第N行数据
					  * Seq(20181201,appName, rawReq,.........................)
					  * Seq(20181201,appName, rawReq,.........................)
					  * Seq(20181201,appName, rawReq,.........................)
					  * )
					  */
					val batchParams = partition.map {
						case (
								(day, appName),
								List(
									rawRequest, effectiveRequest, adRequest,
									isRTB, succRTB, adShow, adClick, expense, cost)
							) => Seq(
									day, appName, rawRequest, effectiveRequest, adRequest,
									isRTB, succRTB, adShow, adClick, expense, cost
							)
					}.toSeq
					sql"insert into r_app_analysis values (?, ?, ?, ?, ?, ?, ?, ?,?,?, ?)".batch(batchParams: _*).apply()
				}
			})


		sc.stop()

	}

}
